# -*- coding: utf-8 -*-

import redis
from HifoEsf.utils.redis_pool import POOL, PROXY_REDIS_KEY
from HifoEsf.utils.redis_case_filter import get_filter_flag
from HifoEsf.items import CommunityItem, CaseItem, BuildingItem, BuildingUnitItem
from HifoEsf.items_gp import GP01Item, GP02Item, GP03Item, GP04Item, GP05Item
from HifoEsf.utils.mongoclient import MongodbClient
from HifoEsf.utils.pcd import PCD_LIST
from dateutil import parser
import json
import datetime
import time
from math import floor
import pandas as pd
import re
import logging
from selenium import webdriver

logger = logging.getLogger(__name__)


class RedisConnPipeline(object):
    def open_spider(self, spider):
        spider.redis_conn = redis.StrictRedis(connection_pool=POOL)
        spider.redis_conn.delete(PROXY_REDIS_KEY)

    def close_spider(self, spider):
        spider.redis_conn.close()


class MongoClientPipeline(object):
    def open_spider(self, spider):
        self.clo = MongodbClient()

    def close_spider(self, spider):
        self.clo.close()

    def process_item(self, item, spider):
        # 房多多
        if spider.name == 'fangduoduo':
            self.clo.set_db('zsq_fangdd')
            if isinstance(item, CommunityItem):
                self.clo.set_col('community')
                self.clo.insert_one(dict(item))
            if isinstance(item, CaseItem):
                self.clo.set_col('case')
                self.clo.insert_one(dict(item))
        # 安居客
        if spider.name == 'anjuke':
            self.clo.set_db('newhours')
            if isinstance(item, CaseItem):
                self.clo.set_col('anjuke_esf_case')
                filter_item = dict()
                filter_item['caseId'] = item['caseId']
                self.clo.update_one(filter_item, dict(item))
        # 安居客楼栋
        if spider.name == 'anjuke_building':
            self.clo.set_db('newhours')
            if isinstance(item, BuildingItem):
                self.clo.set_col('anjuke_esf_building')
                filter_item = dict()
                filter_item['buildingId'] = item['buildingId']
                self.clo.update_one(filter_item, dict(item))
            if isinstance(item, BuildingUnitItem):
                self.clo.set_col('anjuke_esf_unit')
                filter_item = dict()
                filter_item['buildingId'] = item['buildingId']
                filter_item['unitId'] = item['unitId']
                self.clo.update_one(filter_item, dict(item))
        # 贝壳
        if spider.name == 'beike_cj':
            self.clo.set_db('newhours')
            if isinstance(item, CommunityItem):
                self.clo.set_col('beike_esf_community')
                filter_item = dict()
                filter_item['guid'] = item['guid']
                self.clo.update_one(filter_item, dict(item))
            if isinstance(item, CaseItem):
                self.clo.set_col('beike_esf_case')
                filter_item = dict()
                filter_item['caseId'] = item['caseId']
                self.clo.update_one(filter_item, dict(item))
        if spider.name == 'beike_cj_v2':
            self.clo.set_db('newhours')
            if isinstance(item, CaseItem):
                self.clo.set_col('beike_esf_case')
                filter_item = dict()
                filter_item['caseId'] = item['caseId']
                self.clo.update_one(filter_item, dict(item))
        if spider.name == 'beike_building':
            self.clo.set_db('newhours')
            if isinstance(item, BuildingUnitItem):
                self.clo.set_col('beike_esf_unit')
                filter_item = dict()
                filter_item['buildingId'] = item['buildingId']
                filter_item['unitId'] = item['unitId']
                self.clo.update_one(filter_item, dict(item))
        # 房天下
        if spider.name == 'fangtianxia':
            self.clo.set_db('newhours')
            if isinstance(item, CommunityItem):
                self.clo.set_col('fangtianxia_esf_community')
                filter_item = dict()
                filter_item['guid'] = item['guid']
                self.clo.update_one(filter_item, dict(item))
            if isinstance(item, CaseItem):
                self.clo.set_col('fangtianxia_esf_case')
                filter_item = dict()
                filter_item['caseId'] = item['caseId']
                self.clo.update_one(filter_item, dict(item))
        # 爱武隆
        if spider.name == 'aiwulong':
            self.clo.set_db('newhours')
            if isinstance(item, CaseItem):
                self.clo.set_col('esf_case')
                filter_item = dict()
                filter_item['caseId'] = item['caseId']
                self.clo.update_one(filter_item, dict(item))
        # 透明搜房
        if spider.name == 'tmsf':
            self.clo.set_db('newhours')
            if isinstance(item, CommunityItem):
                self.clo.set_col('esf_community')
                filter_item = dict()
                filter_item['guid'] = item['guid']
                self.clo.update_one(filter_item, dict(item))
            if isinstance(item, CaseItem):
                self.clo.set_col('esf_case')
                filter_item = dict()
                filter_item['caseId'] = item['caseId']
                self.clo.update_one(filter_item, dict(item))
        # gpw
        if spider.name == 'gpw':
            self.clo.set_db('gpw')
            if isinstance(item, GP01Item):
                self.clo.set_col('gpw_es_info')
                # self.clo.insert_one(dict(item))
                filter_item = dict()
                filter_item['name'] = item['name']
                filter_item['districtName'] = item['districtName']
                self.clo.update_one(filter_item, dict(item))
            if isinstance(item, GP02Item):
                self.clo.set_col('gpw_es_detail')
                # self.clo.insert_one(dict(item))
                filter_item = dict()
                filter_item['id'] = item['id']
                self.clo.update_one(filter_item, dict(item))
            if isinstance(item, GP03Item):
                self.clo.set_col('gpw_bd')
                # self.clo.insert_one(dict(item))
                filter_item = dict()
                filter_item['id'] = item['id']
                self.clo.update_one(filter_item, dict(item))
            if isinstance(item, GP04Item):
                self.clo.set_col('gpw_type')
                self.clo.insert_one(dict(item))
            if isinstance(item, GP05Item):
                self.clo.set_col('gpw_avg_price')
                # self.clo.insert_one(dict(item))
                filter_item = dict()
                filter_item['id'] = item['id']
                self.clo.update_one(filter_item, dict(item))


class PcdPipeline(object):
    def open_spider(self, spider):
        # 方式一：直接解析list，可以用scrapyd部署
        spider.pcd_df = pd.DataFrame(PCD_LIST)

        # 方式二：读取csv文件，无法使用scrapyd部署
        # spider.pcd_df = pd.read_csv(filepath_or_buffer='HifoEsf/csv/pcd.csv',
        #                             header=0,
        #                             names=['id', 'type', 'name', 'parentId', 'simpleName', 'otherName'],
        #                             encoding='utf-8', )

    def close_spider(self, spider):
        del spider.pcd_df


# 房多多
class FangduoduoPipeline(object):
    def process_item(self, item, spider):
        if spider.name == 'fangduoduo':
            # str空值处理
            for item_tup in item.items():
                item[item_tup[0]] = self.deal_item(item_tup[1])
            # transportFacility字段处理
            item['transportFacility'] = self.deal_transport_facility(item['transportFacility'])
            # 小区字段处理
            if isinstance(item, CommunityItem):
                # updateDate字段处理
                item['updateDate'] = self.timestr_to_isodate(item['updateDate'])
                # propertyType字段处理
                item['type'] = self.list_to_str(item['type'])
                # totalRoomCount字段处理
                item['totalRoomCount'] = self.str_to_int(item['totalRoomCount'])
            #  案例字段处理
            if isinstance(item, CaseItem):
                # caseDate字段新增
                item['caseDate'] = int(datetime.datetime.now().strftime("%Y%m%d"))
                # caseMonth字段新增
                item['caseMonth'] = int(datetime.datetime.now().strftime("%Y%m"))
                # caseType字段新增
                item['caseType'] = '挂牌'
                # updateDate字段处理
                item['updateDate'] = self.stamp_to_isodate(item['updateDate'])
                # createDate字段处理
                item['createDate'] = self.stamp_to_isodate(item['createDate'])
                # totalPrice字段处理
                item['totalPrice'] = self.deal_total_price(item['totalPrice'])
                # buildArea字段处理
                item['buildArea'], item['stdBuildAreaInt'], item['stdBuildArea'] = self.deall_buid_area(
                    item['buildArea'])
            item['fetchDate'] = self.stamp_to_isodate(int(time.time() * 10 ** 3))
            item['sr'] = 'Fangdd'
            item['siteName'] = '房多多'
        return item

    @staticmethod
    def deal_item(item_val):
        """
        小区&案例字段str空值处理
        :param item_val: 字段的value值
        :return:
        """
        try:
            assert isinstance(item_val, str)
            item_val = item_val.strip() if item_val.strip() else None
        except:
            return item_val
        else:
            return item_val

    @staticmethod
    def timestr_to_isodate(time_str):
        """
        日期类型转换 str -> ISODate
        :param time_str:
        :return:
        """
        try:
            assert isinstance(time_str, str)
            iso_date = parser.parse(time_str)
        except AssertionError:
            return
        except:
            return time_str
        else:
            return iso_date

    @staticmethod
    def list_to_str(_li):
        """
        列表转字符串
        :param _li:
        :return:
        """
        try:
            assert isinstance(_li, list)
            ret = ','.join(_li)
        except:
            return _li
        else:
            return ret

    @staticmethod
    def deal_transport_facility(_li):
        """
        交通状况字段反序列化
        :param _li:
        :return:
        """
        try:
            assert len(_li)
            ret_li = list()
            for ret_dict in _li:
                ret_dict.pop('__typename')
                ret_li.append(ret_dict)
            ret_str = json.dumps(ret_li, ensure_ascii=False)
        except:
            return
        else:
            return ret_str

    def stamp_to_isodate(self, stamp):
        """
        将时间戳转换为ISODate
        :param stamp: 1581782400000
        :return:
        """
        try:
            date_obj = datetime.datetime.fromtimestamp(int(stamp) / 1000, None)
            date_str = date_obj.strftime("%Y-%m-%dT%H:%M:%S")
            iso_date = self.timestr_to_isodate(date_str)
        except:
            return stamp
        else:
            return iso_date

    @staticmethod
    def deal_total_price(price):
        """
        案例总价字段处理
        :param price:
        :return:
        """
        try:
            assert (isinstance(price, int) or isinstance(price, float))
            ret = float(price)
        except:
            return price
        else:
            return ret

    @staticmethod
    def deall_buid_area(area):
        """
        案例建筑面积字段处理
        :param area:
        :return:
        """
        try:
            assert (isinstance(area, int) or isinstance(area, float))
            ret_float = float(area)
            ret_int = floor(ret_float)
            ret_str = str(ret_int)
        except:
            return area, None, None
        else:
            return ret_float, ret_int, ret_str

    @staticmethod
    def str_to_int(_str):
        """
        字符串转整型
        :param _str:
        :return:
        """
        regex = re.compile(r'\d+')
        if not isinstance(_str, int):
            try:
                ret_li = regex.findall(_str)
                assert len(ret_li) == 1
                ret = int(ret_li[0])
            except:
                return
            else:
                return ret
        else:
            return _str


# 安居客
class AnjukePipeline(object):
    def open_spider(self, spider):
        if spider.name == 'anjuke':
            case_id_filter = spider.settings['CASE_ID_FILTER'] if spider.settings['CASE_ID_FILTER'] != None else True
            if case_id_filter:
                logger.info(f'>>>>{spider.name}:案例id去重开启<<<<')
                # todo 连接mongo集群需优化
                col_name_li: list = ['anjuke_esf_case', ]
                spider.filter_flag = get_filter_flag('newhours', col_name_li, 'anjuke_esf_case_id',
                                                     mongo_client=MongodbClient()._client)
                logger.info(f'>>>>{spider.name}:案例id去重配置成功<<<<') if spider.filter_flag else logger.error(
                    f'>>>>{spider.name}:案例id去重配置失败<<<<')
            else:
                spider.filter_flag = None
                logger.info(f'>>>>{spider.name}:不根据案例id进行去重<<<<')

    def process_item(self, item, spider):
        if spider.name == 'anjuke':
            # str空值处理
            for item_tup in item.items():
                item[item_tup[0]] = self.deal_item(item_tup[1])
            if isinstance(item, CaseItem):
                # totalPrice字段处理
                item['totalPrice'] = self.deal_total_price(item['totalPrice'])
                # caseDate字段新增
                item['caseDate'] = int(datetime.datetime.now().strftime("%Y%m%d"))
                # caseMonth字段新增
                item['caseMonth'] = int(datetime.datetime.now().strftime("%Y%m"))
                # caseType字段新增
                item['caseType'] = '挂牌'
            item['fetchDate'] = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            item['updateDate'] = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            item['sr'] = 'Anjuke'
            item['siteName'] = '安居客'
        return item

    @staticmethod
    def deal_item(item_val):
        """
        小区&案例字段str空值处理
        :param item_val: 字段的value值
        :return:
        """
        try:
            assert isinstance(item_val, str)
            item_val = item_val.strip() if item_val.strip() else None
        except:
            return item_val
        else:
            return item_val

    @staticmethod
    def deal_total_price(price):
        """
        案例总价字段处理
        :param price:
        :return:
        """
        try:
            assert isinstance(price, list)
            ret = ''.join([i.strip() for i in price if i and i.strip()])
        except:
            return price
        else:
            return ret


# 安居客楼盘
class AnjukeLpPipeline(object):
    def open_spider(self, spider):
        if spider.name == 'anjuke_loupan':
            spider.clo = MongodbClient()
            chrome_options = webdriver.ChromeOptions()
            chrome_options.add_experimental_option('debuggerAddress', '127.0.0.1:5247')
            spider.driver = webdriver.Chrome(options=chrome_options)

    def close_spider(self, spider):
        spider.clo.close()


# 安居客楼栋
class AnjukeBdPipeline(object):
    def open_spider(self, spider):
        if spider.name == 'anjuke_building':
            spider.clo = MongodbClient()

    def close_spider(self, spider):
        spider.clo.close()

    def process_item(self, item, spider):
        if spider.name == 'anjuke_building':
            item['fetchDate'] = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            item['updateDate'] = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            item['sr'] = 'Anjuke'
            item['siteName'] = '安居客'
        return item


# 贝壳
class BiekeCjPipeline(object):
    def process_item(self, item, spider):
        if spider.name == 'beike_cj':
            # str空值处理
            for item_tup in item.items():
                item[item_tup[0]] = self.deal_item(item_tup[1])
            if isinstance(item, CommunityItem):
                # other字段处理
                item['other'] = self.list_to_str(item['other'])
            item['fetchDate'] = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            item['updateDate'] = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            item['sr'] = 'Beike'
            item['siteName'] = '贝壳'
        if spider.name == 'beike_cj_v2':
            if not item['totalFloor']:
                del item['totalFloor']
            item['fetchDate'] = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            item['updateDate'] = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            item['sr'] = 'Beike'
            item['siteName'] = '贝壳'
        return item

    @staticmethod
    def deal_item(item_val):
        """
        小区&案例字段str空值处理
        :param item_val: 字段的value值
        :return:
        """
        try:
            assert isinstance(item_val, str)
            item_val = item_val.strip() if item_val.strip() else None
        except:
            return item_val
        else:
            return item_val

    @staticmethod
    def list_to_str(_li):
        """
        列表转字符串
        :param _li:
        :return:
        """
        try:
            assert isinstance(_li, list)
            ret = ''.join([i.strip() for i in _li if i and i.strip()])
        except:
            return _li
        else:
            return ret


# 贝壳楼栋
class BeikeBdPipeline(object):
    def open_spider(self, spider):
        if spider.name == 'beike_building':
            spider.clo = MongodbClient()

    def close_spider(self, spider):
        spider.clo.close()

    def process_item(self, item, spider):
        if spider.name == 'beike_building':
            item['fetchDate'] = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            item['updateDate'] = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            item['sr'] = 'Beike'
            item['siteName'] = '贝壳'
        return item


# 房天下
class FangtianxiaPipeline(object):
    def open_spider(self, spider):
        if spider.name == 'fangtianxia':
            case_id_filter = spider.settings['CASE_ID_FILTER'] if spider.settings['CASE_ID_FILTER'] != None else True
            if case_id_filter:
                logger.info(f'>>>>{spider.name}:案例id去重开启<<<<')
                # todo 连接mongo集群需优化
                col_name_li: list = ['fangtianxia_esf_case', ]
                spider.filter_flag = get_filter_flag('newhours', col_name_li, 'fangtianxia_esf_case_id',
                                                     mongo_client=MongodbClient()._client)
                logger.info(f'>>>>{spider.name}:案例id去重配置成功<<<<') if spider.filter_flag else logger.error(
                    f'>>>>{spider.name}:案例id去重配置失败<<<<')
            else:
                spider.filter_flag = None
                logger.info(f'>>>>{spider.name}:不根据案例id进行去重<<<<')

    def process_item(self, item, spider):
        if spider.name == 'fangtianxia':
            # str空值处理
            for item_tup in item.items():
                item[item_tup[0]] = self.deal_item(item_tup[1])
            if isinstance(item, CommunityItem):
                # other字段处理
                if isinstance(item['other'], list):
                    item['other'] = [i.strip() for i in item['other'] if i and i.strip()]
            if isinstance(item, CaseItem):
                # isElevator字段处理
                if isinstance(item['isElevator'], str):
                    item['isElevator'] = True if '有' in item['isElevator'] else False
                else:
                    item['isElevator'] = None
                # buildStructure字段处理
                item['buildStructure'] = self.list_to_str(item['buildStructure'])
                # buildType字段处理
                item['buildType'] = self.list_to_str(item['buildType'])
                # calQuality字段处理
                item['calQuality'] = self.list_to_str(item['calQuality'])
                # other字段处理
                item['other'] = self.list_to_str(item['other'])
                # property字段处理
                item['property'] = self.list_to_str(item['property'])
            item['fetchDate'] = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            item['updateDate'] = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            item['sr'] = 'Fangtianxia'
            item['siteName'] = '房天下'
        return item

    @staticmethod
    def deal_item(item_val):
        """
        小区&案例字段str空值处理
        :param item_val: 字段的value值
        :return:
        """
        try:
            assert isinstance(item_val, str)
            item_val = item_val.strip() if item_val.strip() else None
        except:
            return item_val
        else:
            return item_val

    @staticmethod
    def list_to_str(_li):
        """
        列表转字符串
        :param _li:
        :return:
        """
        try:
            assert isinstance(_li, list)
            ret = ''.join([i.strip() for i in _li if i and i.strip()])
        except:
            return _li
        else:
            return ret


# 爱武隆
class AiwulongPipeline(object):
    def process_item(self, item, spider):
        if spider.name == 'aiwulong':
            # str空值处理
            for item_tup in item.items():
                item[item_tup[0]] = self.deal_item(item_tup[1])
            if isinstance(item, CaseItem):
                # listingDate字段处理
                item['listingDate'] = self.deal_date(item['listingDate'])
                # createDate字段处理
                item['createDate'] = self.deal_date(item['createDate'])
            item['fetchDate'] = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            item['sr'] = 'Aiwulong'
            item['siteName'] = '爱武隆'
        return item

    @staticmethod
    def deal_item(item_val):
        """
        小区&案例字段str空值处理
        :param item_val: 字段的value值
        :return:
        """
        try:
            assert isinstance(item_val, str)
            item_val = item_val.strip() if item_val.strip() else None
        except:
            return item_val
        else:
            return item_val

    @staticmethod
    def deal_date(date: str):
        """
        日期处理
        :param date:
        :return:
        """
        if date and any(['分钟' in date, '小时' in date, '天' in date]):
            num = int(re.findall('\d+', date)[0])
            if '分钟' in date:
                ret = datetime.datetime.now().strftime('%Y-%m-%d')
            elif '小时' in date:
                ret = datetime.datetime.now() + datetime.timedelta(hours=-num)
                ret = ret.strftime('%Y-%m-%d')
            else:
                ret = datetime.datetime.now() + datetime.timedelta(days=-num)
                ret = ret.strftime('%Y-%m-%d')
        else:
            ret = date
        return ret


# 透明搜房
class TmsfPipeline(object):
    def process_item(self, item, spider):
        if spider.name == 'tmsf':
            # str空值处理
            for item_tup in item.items():
                item[item_tup[0]] = self.deal_item(item_tup[1])
            if item['cityName'] == '杭州':
                if isinstance(item, CommunityItem):
                    item['updateDate'] = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                    # totalRoomCount字段处理
                    item['totalRoomCount'] = item['totalRoomCount'] if item['totalRoomCount'] else None
            item['fetchDate'] = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            item['sr'] = 'Tmsf'
            item['siteName'] = '透明售房网'
        return item

    @staticmethod
    def deal_item(item_val):
        """
        小区&案例字段str空值处理
        :param item_val: 字段的value值
        :return:
        """
        try:
            assert isinstance(item_val, str)
            item_val = item_val.strip() if item_val.strip() else None
        except:
            return item_val
        else:
            return item_val
